package com.zy.flink.dataset.datastream;


import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Map;
import java.util.concurrent.TimeUnit;


/**
 * @author Dell
 * @Title:
 * @Description: 数据源批处理，用的是实时流api
 * @date 2024/1/18
 */
public class kafkaAndCkSourceMain {
    public static void main(String[] args) {

        // 创建Flink执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.of(10, TimeUnit.SECONDS)));

        // 数据源，读clickhouse，防止重复处理此处并行度1，按需设置
        // 读kafka可注意算子链，即source和sink并行度一致
        DataStream<Map<String, Object>> users = env.addSource(new ClickHouseSource()).setParallelism(1);

        // transformation: map keyBy timeWindow apply

        users.print("test");
        // execute
        try {
            env.execute();
        } catch (Exception e) {
            System.out.println(e.getMessage());
        }



    }
}
